Back to Main Menu

Data Exchange

Example Process for Automating Data Upload

The following Python scripts illustrate how a process can be setup to check an upload folder for new uploads and manage the upload via Data Exchange, including checks for upload errors, using the API endpoint POST /api/v2/DataExchangeJob.

 

  1. """
  2.     Example script to automate data exchange (Assetic.DataExchangeProcessNew.py)
  3.     First of two scripts
  4.     This script uploads to Assetic the files in the 'ToLoad' folder.  The
  5.     filename is parsed to get the asset category which allows the data exchange
  6.     profile to be determined.  A data exchange task is created for the document
  7.     The name of the file is changed to the task id and moved to the 'InProgress'
  8.     folder.  This allows the subsequent script to identify the task and check
  9.     progress
  10. """
  11. import assetic
  12. import os
  13. import shutil
  14. import sys
  15. import base64
  16.   
  17. # Assetic SDK instance
  18. asseticsdk = assetic.AsseticSDK("C:/Users/kwilton/assetic.ini",None,"Info")
  19. ##API instances
  20. # Document API
  21. docapi = assetic.DocumentApi()
  22. # Data Exchange API
  23. dataexchangeapi = assetic.DataExchangeJobApi()
  24.   
  25. # File Processing Directory structure
  26. intdir =  "C:/temp/Integration/"
  27. sourcedir = intdir +  "ToLoad/"
  28. inprogress = intdir +  "InProgress/"
  29. errordir = intdir +  "Error/"
  30. processeddir = intdir +  "Processed/"
  31.   
  32. ##preview
  33. profiles =  {'Sewer Nodes':  'c7e6d3ed-3917-e611-9458-06edd62954d7',
  34.   'Roads':  'd889d99b-3317-e611-9458-06edd62954d7'}
  35. ##demo
  36. profiles =  {'Sewer Nodes':  'c7e6d3ed-3917-e611-9458-06edd62954d7',
  37.   'Roads':  '876a0a4e-c732-e611-945f-06edd62954d7',
  38.   'Water Pressure Pipes':  '3abd431c-2017-e611-9812-0632cf3be881'}
  39.   
  40. ##Loop through files in folder
  41. files = os.listdir(sourcedir)
  42. for filename in files:
  43.     fullfilename = sourcedir + filename
  44.   #get category, assumes file name has a whitespace between date and category
  45.     filedate, categorywithext = filename.split(None,1)
  46.     category, fileext = os.path.splitext(categorywithext)
  47.     asseticsdk.logger.info("File Date: {0}, Category: {1}".format(
  48.         filedate, category))
  49.   
  50.   ##read file and encode for upload
  51.   with open(fullfilename,  "rb")  as f:
  52.         data = f.read()
  53.   if sys.version_info <  (3,0):
  54.         filecontents = data.encode("base64")
  55.   else:
  56.         filecontents = base64.b64encode(data)
  57.         filecontents = filecontents.decode(encoding="utf-8", errors="strict")
  58.   
  59.   ##create file properties object and set values 
  60.     file_properties = \
  61.         assetic.Assetic3IntegrationRepresentationsFilePropertiesRepresentation()
  62.     file_properties.name = filename
  63.     file_properties.mimetype =  'csv'
  64.     file_properties.filecontent = filecontents
  65.     filearray =  [file_properties]
  66.   ##create document object and assign values, including file properties
  67.     document = \
  68.         assetic.Assetic3IntegrationRepresentationsDocumentRepresentation()
  69.     document.label =  'Data Exchange - '  + filename
  70.     document.file_extension =  'csv'
  71.     document.document_type =  'DataExchange'
  72.     document.mime_type =  'csv'
  73.     document.doc_group_id =  1
  74.     document.file_property = filearray
  75.   
  76.   #Upload document and get ID from response
  77.     doc = docapi.document_post(document)
  78.     docid = doc[0].get('Id')
  79.   
  80.   #get dataexchange profile for category
  81.     profileid = profiles.get(category)
  82.   
  83.   ##prepare data exchange parameters
  84.     job = assetic.Assetic3IntegrationRepresentationsDataExchangeJobRepresentation()
  85.     job.profile_id = profileid
  86.     job.document_id = docid
  87.     asseticsdk.logger.info("Profile ID:{0}".format(job.profile_id))
  88.     asseticsdk.logger.info("Job Document ID:{0}".format(job.document_id))
  89.   try:
  90.         task = dataexchangeapi.data_exchange_job_post(job)
  91.   #move file to in-progress, and rename to task id to allow checks
  92.   #on success/failure
  93.         destination = inprogress + task +  '.csv'
  94.         shutil.move(fullfilename,destination)
  95.   except assetic.rest.ApiException  as e:
  96.   #gross error - move to error folder??
  97.         asseticsdk.logger.error('Status {0}, Reason: {1}'.format(
  98.             e.status,e.reason))
  99.         destination = errordir + task +  '.csv'
  100.         shutil.move(fullfilename,destination)
  101.

How it works

The folder locations are defined in the script.  It assumes that an external process is populating the source folder with files that are to be uploaded.

##File Processing Directory structure intdir =  "C:/temp/Integration/" sourcedir = intdir +  "ToLoad/" inprogress = intdir +  "InProgress/" errordir = intdir +  "Error/" processeddir = intdir +  "Processed/"

Each file in the source folder will be processed one at a time

files = os.listdir(sourcedir)  for filename in files:

 In this example each Asset Category will have a separate Date Exchange profile, so the profile ID for each category is defined

##List of categories and associated dataexchange profile profiles =  {'Sewer Nodes':  'c7e6d3ed-3917-e611-9458-06edd62954d7',  'Roads':  '876a0a4e-c732-e611-945f-06edd62954d7',  'Water Pressure Pipes':  '3abd431c-2017-e611-9812-0632cf3be881'}

 The filename is parsed to determine the category and hence the profile ID

  #get category, assumes file name has a whitespace between date and category     filedate, categorywithext = filename.split(None,1)     category, fileext = os.path.splitext(categorywithext)

 Get the dataexchange profile for the category

    profileid = profiles.get(category)

 The document is uploaded and the document ID obtained

  #Upload document and get ID from response     doc = docapi.document_post(document)     docid = doc[0].get('Id')

 

After preparing the job information the job is posted to Data Exchange.  If there is an exception returned it means there was a problem with creating the job.  Perhaps the profile ID is incorrect or there are insufficient priviledges to create the job.

If there is no exception the Data Exchange task ID is returned.  Since tasks are queued it may not run immediately.  To allow the task to be checked at a later stage the file is renamed to the task ID and moved to the 'inprogress' folder. 

  try:         task = dataexchangeapi.data_exchange_job_post(job)  #move file to in-progress, and rename to task id to allow checks  #on success/failure         destination = inprogress + task +  '.csv'         shutil.move(fullfilename,destination)  except assetic.rest.ApiException  as e:  #gross error - move to error folder         destination = errordir + task +  '.csv'         shutil.move(fullfilename,destination)         asseticsdk.logger.error('Status {0}, Reason: {1}'.format(e.status,e.reason))

Checking if a task has processed

The following Python script uses the file name to check to see if the data exchange task has run.  If it has run successfully then the file is moved to the 'Processed' folder, otherwise it is moved to the 'Error' folder.  This script is scheduled to run periodically.

 

  1. """
  2.     Example script to automate data exchange (Assetic.DataExchangeCheckTask.py)
  3.     Second of two scripts
  4.     This script checks the 'InProcess' folder to see if task has beeen
  5.     processed in data exchange. The names of the files in this folder have been
  6.     changed from the original name to the data exchange task id
  7.     If a task has been processed then it moves the file to either the
  8.     'Processed' folder or the 'Error' folder.  If there is an error
  9.     the error file generated by data exchange is also downloaded and
  10.     placed in the error folder
  11. """
  12. import assetic
  13. import os
  14. import shutil
  15. import sys
  16.   
  17. ##Assetic SDK instance    
  18. asseticsdk = assetic.AsseticSDK("c:/users/you/assetic.ini",None,"Info")
  19. ##create an instance of the DataExchange API
  20. dataexchangeapi = assetic.DataExchangeTaskApi()
  21. ##Create an instance for document get.
  22. ##Use sdk client for document get.  Returns header 'content-disposition'
  23. ##The header has the file name of the doc
  24. #Document  API
  25. docapi = assetic.DocumentApi(asseticsdk.client_for_docs)
  26.   
  27. # Define file path for output report files
  28. intdir =  "C:/temp/Integration/"
  29. sourcedir = intdir +  "ToLoad/"
  30. inprogressdir = intdir +  "InProgress/"
  31. errordir = intdir +  "Error/"
  32. processeddir = intdir +  "Processed/"
  33.   
  34. ##define function for getting error file
  35. def getfile(docid,docapi,filedir):
  36.   try:
  37.         getfile = docapi.document_get_document_file(docid)
  38.   except assetic.rest.ApiException  as e:
  39.         asseticsdk.logger.error('Status {0}, Reason: {1}'.format(
  40.             e.status,e.reason))
  41.   
  42.   if getfile !=  None:
  43.   if  'attachment'  in getfile[1]  and  'filename='  in getfile[1]:
  44.             filename = getfile[1].split('filename=',1)[1]
  45.   if  '"'  in filename or  "'"  in filename:
  46.                 filename = filename[1:-1]
  47.             fullfilename = filedir + filename
  48.   else:
  49.             fullfilename = filedir +filename 
  50.   
  51.         data = getfile[0]
  52.   if sys.version_info >=  (3,0):
  53.   try:
  54.                 data = data.encode('utf-8')
  55.   except:
  56.                 data = data
  57.   with open( fullfilename,  'wb'  )  as out_file:
  58.             out_file.write(data)
  59.             asseticsdk.logger.info('Created file: {0}'.format(fullfilename))
  60. ##end function
  61.   
  62. files = os.listdir(inprogressdir)
  63. for filename in files:
  64.     fullfilename = inprogressdir + filename
  65.     taskguid, fileext = os.path.splitext(filename)
  66.   
  67.   try:
  68.         task = dataexchangeapi.data_exchange_task_get(taskguid)
  69.   except assetic.rest.ApiException  as e:
  70.         asseticsdk.logger.error('Status {0}, Reason: {1}'.format(
  71.             e.status,e.reason))
  72.   else:
  73.         status = task.get('StatusDescription')
  74.   if status ==  'Complete with error':
  75.   #move file to error
  76.             destination = errordir + filename
  77.             shutil.move(fullfilename,destination)
  78.   ##also get error file
  79.             errordocid = task.get('ErrorDocumentId')
  80.             errordesc = task.get('Summary')
  81.   if errordocid ==  None:
  82.                 asseticsdk.logger.info("Error Description: {0}".format(
  83.                     errordesc))
  84.   else:
  85.                 asseticsdk.logger.info("Error document: {0}".format(errordocid))
  86.                 errordirtask = errordir + taskguid
  87.                 getfile(errordocid,docapi,errordirtask)
  88.   
  89.   elif status ==  'Completed':
  90.   #move file to Completed
  91.             destination = processeddir + filename
  92.             shutil.move(fullfilename,destination)  
  93.

How it works

The script loops through files in the inprogress folder and obtains the Data Exchange taskid using the file name of the file.  This script is schedule to run periodically.

files = os.listdir(inprogressdir)  for filename in files:     fullfilename = inprogressdir + filename     taskguid, fileext = os.path.splitext(filename)

The task is requested and status of the response checked

  try:         task = dataexchangeapi.data_exchange_task_get(taskguid)  except assetic.rest.ApiException  as e:         asseticsdk.logger.error('Status {0}, Reason: {1}'.format(e.status,e.reason))  else:         status = task.get('StatusDescription')

If the status is 'Complete with Error' the file is moved to the error folder.  If there error is due to an error with field mapping the the 'Summary' describes the problem.

If there is a data error then an error file is generated by Data Exchange.  This file is downloaded and saved in the 'error' folder along with the document.  The function 'getfile' is used to get the file. 

#move file to error destination = errordir + filename shutil.move(fullfilename,destination)  ##also get error file errordocid = task.get('ErrorDocumentId') errordesc = task.get('Summary')  if errordocid ==  None:    asseticsdk.logger.info(errordesc)  else:    asseticsdk.logger.info(errordocid)    errordirtask = errordir + taskguid    getfile(errordocid,docapi,errordirtask)

The function 'getfile' is defined in the script body.  Defining functions within the script can make it easier to read the script and perform repetitive tasks

##define function for getting error file  def getfile(docid,docapi,filedir):     getfile = docapi.document_get_document_file(docid)  if getfile !=  None:  if  'attachment'  in getfile[1]  and  'filename='  in getfile[1]:             filename = getfile[1].split('filename=',1)[1]  if  '"'  in filename or  "'"  in filename:                 filename = filename[1:-1]             fullfilename = filedir + filename         else:             fullfilename = filedir +filename           data = getfile[0]  if sys.version_info >=  (3,0):  try:                 data = data.encode('utf-8')  except:                 data = data         with open( fullfilename,  'wb'  )  as out_file:             out_file.write(data)             asseticsdk.logger.error ('Created file: {0}'.format(fullfilename))  ##end function

If the status is 'completed' it means the Data Exchange task has completed without error.  The file is moved to the completed folder and no further processing on that file takes place.

elif status ==  'Completed':  #move file to Completed    destination = processeddir + filename    shutil.move(fullfilename,destination)

If the task has not been processed then the file remains in the 'inprogress' folder.